#!/usr/bin/perl

package MigrationFen;

use strict;
use warnings;

use File::Find;
use File::Basename;
use JSON qw(decode_json);
use POSIX qw(strftime);
use DBI;

use Const;
use Log;
use Utils qw(dd);
use Export;

use utf8;
use open qw(:std :utf8);
binmode STDOUT, ":utf8";

sub list
{
    my $migration_name = shift;

    my $migration_path = $Const::CONF_PATH . $Const::DS . "migrations-fen";
    my %migrations;
    find(sub {
        if (/\.json$/ && -f $_) {
            my $filename = $_;
            my $basename = basename($File::Find::name);
            if (defined $migration_name) {
                if ("${migration_name}.json" eq $basename or "${migration_name}" eq $basename) {
                    $migrations{$basename} = $filename;
                    last;
                }
            } else {
                $migrations{$basename} = $filename;
            }
        }
    }, $migration_path);

    Log::info("fensheng migration list:");

    for my $key (sort keys %migrations) {
        my $filename = $migrations{$key};
        my $json = Utils::readFile($migration_path. $Const::DS. $filename);
        my $migration = decode_json($json);
        Log::info("migration: $key");

        my $is_com_table = defined $migration->{is_com_table} ? $migration->{is_com_table} : 0;
        my $is_long_comid = defined $migration->{is_long_comid}? $migration->{is_long_comid} : 0;
        Log::info("    table: $migration->{table}, is_com_table: ${is_com_table}, is_long_comid: ${is_long_comid}");

        $migration->{"src_conn"} = _getConn($Const::CONF_PATH. $Const::DS. "conns" . $Const::DS . $migration->{src_conn_name} . ".json");
        Log::info("    src conn cmd : [ gsql -h $migration->{src_conn}->{host} -p $migration->{src_conn}->{port} -d $migration->{src_conn}->{database} -U $migration->{src_conn}->{user} -W'$migration->{src_conn}->{pass}' -ar ]");
    }
}

sub connList
{
    my $conns_path = $Const::CONF_PATH. $Const::DS. "conns-fen";
    my %conns;
    find(sub {
        if (/\.json$/ && -f $_) {
            my $filename = $_;
            my $basename = basename($File::Find::name);
            $conns{$basename} = $filename;
        }
    }, $conns_path);

    Log::info("fensheng conn list: ");
    for my $key (sort keys %conns) {
        my $conn = _getConn($conns_path . $Const::DS . $conns{$key});
        Log::info("conn: $key. host: $conn->{host}, port: $conn->{port}, database: $conn->{database}, user: $conn->{user}, pass: $conn->{pass}");
        Log::info("gsql: gsql -h $conn->{host} -p $conn->{port} -U $conn->{user} -d $conn->{database} -W'$conn->{pass}' -ar")
    }
}

sub _getConn
{
    my $filename = shift;
    my $json = Utils::readFile($filename);
    my $conn = decode_json($json);
    return $conn;
}

sub _getFenConn
{
    my $filename = shift;
    my $json = Utils::readFile($filename);
    my $config = decode_json($json);
    return $config;
}

sub run
{
    my $migration_name = shift;

    my $migration_path = $Const::CONF_PATH. $Const::DS. "migrations-fen";
    my %migrations;

    find(sub {
        if (/\.json$/ && -f $_) {
            my $filename = $_;
            my $basename = basename($File::Find::name);

            if (defined $migration_name) {
                if ("$migration_name.json" eq $basename or "$migration_name" eq $basename) {
                    $migrations{$basename} = $filename;
                    last;
                }
            } else {
                $migrations{$basename} = $filename;
            }
        }
    }, $migration_path);

    my $max_processes = getMaxProcesses();
    my @pids;
    my $index = 0;
    my $count = scalar keys %migrations;

    Log::info("Max processes: $max_processes, count: $count");

    for my $key (sort keys %migrations) {
        $index = $index + 1;

        my $filename = $migrations{$key};
        my $json = Utils::readFile($migration_path . $Const::DS . $filename);
        my $migration = decode_json($json);

        if (scalar(@pids) >= $max_processes) {
            my $pid = waitpid(-1, 0);
            if ($pid > 0) {
                @pids = grep { $_ != $pid } @pids;
                $index = $index - 1;
                redo;
            } else {
                die "waitpid failed: $!";
            }
        } else {
            my $pid = fork();
            if ($pid == 0) {
                Log::info("pid: $$, migration_name: $key");
                Log::primary(">>>>>>>> Migration file [$index/$count]: $key. <<<<<<<<");
                _compare("${Const::CONF_PATH}${Const::DS}migrations-fen${Const::DS}${key}");
                exit 0;
            } else {
                push @pids, $pid;
            }
        }
    }

    for my $pid (@pids) {
        waitpid($pid, 0);
    }
}

sub _compare
{
    my $migration_file = shift;

    # migration info
    my $migration_path = $Const::CONF_PATH. $Const::DS. "migrations-fen";
    my $json = Utils::readFile($migration_file);
    my $migration = decode_json($json);
    $migration->{"src_conn"} = _getConn($Const::CONF_PATH. $Const::DS. "conns" . $Const::DS . $migration->{src_conn_name} . ".json");

    Log::info("migration info: ");
    Log::info("    table        : $migration->{table}, is_com_table: $migration->{is_com_table}");
    Log::info("    src conn cmd : [ gsql -h $migration->{src_conn}->{host} -p $migration->{src_conn}->{port} -d $migration->{src_conn}->{database} -U $migration->{src_conn}->{user} -W'$migration->{src_conn}->{pass}' -ar ]");
    Log::info("    sql          : $migration->{sql}");

    # 原表数据库连接
    my $src_dbh = DBI->connect("dbi:Pg:dbname=$migration->{src_conn}->{database};host=$migration->{src_conn}->{host};port=$migration->{src_conn}->{port}", $migration->{src_conn}->{user}, $migration->{src_conn}->{pass}, {
        RaiseError => 1,
        AutoCommit => 1,
    }) or die "Failed to connect to PostgreSQL database: $DBI::errstr";

    # 查询原表列
    my @numerical_columns;
    my @character_columns;
    my $src_table = $migration->{table};
    if (defined $migration->{src_table} && $migration->{src_table} ne "") {
        $src_table = $migration->{src_table};
    }
    my $src_columns = $src_dbh->selectall_arrayref("SELECT column_name, data_type FROM information_schema.columns WHERE table_schema = '$migration->{src_conn}->{schema}' AND table_name = lower('${src_table}') ORDER BY ordinal_position");
    for my $column (@$src_columns) {
        my ($column_name, $data_type) = @$column;
        # Log::info("    Column: $column_name, Type: $data_type");
        if ($data_type =~ /int|numeric|decimal|float|double|money|smallint|bigint/) {
            push @numerical_columns, $column_name;
        } elsif ($data_type =~ /varchar|char|text/) {
            push @character_columns, $column_name;
        }
    }
    if (scalar @{$migration->{numerical_columns}} != 0) {
        @numerical_columns = @{$migration->{numerical_columns}};
    }
    if (scalar @{$migration->{character_columns}}!= 0) {
        @character_columns = @{$migration->{character_columns}};
    }
    Log::info("    Numerical columns: [ ". join(", ", @numerical_columns). " ]");
    Log::info("    Character columns: [ ". join(", ", @character_columns). " ]");

    my $conns_fen_path = $Const::CONF_PATH. $Const::DS. "conns-fen";
    my %coms;
    find(sub {
        if (/\.json$/ && -f $_) {
            my $filename = $_;
            my $parsename = fileparse(basename($File::Find::name), ".json");
            my $fen_conn = _getFenConn($filename);
            
            $coms{$parsename} = $fen_conn;
        }
    }, $conns_fen_path);

    # 循环 comids
    for my $comid (keys %coms) {
        Log::info("Comid: $comid, gsql command: [ gsql -h $coms{$comid}->{host} -p $coms{$comid}->{port} -d $coms{$comid}->{database} -U $coms{$comid}->{user} -W'$coms{$comid}->{pass}' -ar ]");
        
        my $dst_dbh = DBI->connect("dbi:Pg:dbname=$coms{$comid}->{database};host=$coms{$comid}->{host};port=$coms{$comid}->{port}", $coms{$comid}->{user}, $coms{$comid}->{pass}, {
            RaiseError => 1,
            AutoCommit => 1,
        });
        
        # 查询原表count
        my $src_sql = sprintf(_getSQL($migration, $comid, "src"), " COUNT(1) AS count ", $migration->{src_conn}->{schema});
        Log::info("    Select src count sql: $src_sql");
        my $src_count = $src_dbh->selectrow_array($src_sql);

        my $dst_sql = sprintf(_getSQL($migration, $comid, "dst"), " COUNT(1) AS count ", $coms{$comid}->{schema});
        Log::info("    Select dst count sql: $dst_sql");
        my $dst_count = $dst_dbh->selectrow_array($dst_sql);
        if ($src_count!= $dst_count) {
            Log::error("    src count: $src_count, dst count: $dst_count miss match!");
        } else {
            Log::info("    src count: $src_count, dst count: $dst_count match.");
        }

        if ($dst_count == 0 && $src_count == 0) {
            Log::warn("    Warning!!! [${migration_file}] [${comid}] src count and dst count is 0 ! Require attention.");
        }
        
        # 循环 numerical_columns
        my @diff_numerical_result;
        for my $column (@numerical_columns) {
            Log::info("    Numerical column: $column");
            my $src_sql = sprintf(_getSQL($migration, $comid, "src"), " SUM($column) AS sum ", $migration->{src_conn}->{schema});
            Log::info("        Select src sum sql: $src_sql");
            my $src_sum = $src_dbh->selectrow_array($src_sql);
            unless (defined $src_sum) {
                $src_sum = 0;
            }

            my $dst_sql = sprintf(_getSQL($migration, $comid, "dst"), " SUM($column) AS sum ", $coms{$comid}->{schema});
            Log::info("        Select dst sum sql: $dst_sql");
            my $dst_sum = $dst_dbh->selectrow_array($dst_sql);
            unless (defined $dst_sum) {
                $dst_sum = 0;
            }
            
            if ($src_sum != $dst_sum) {
                Log::error("        src [${column}] sum: $src_sum, dst sum: $dst_sum miss match!");
            } else {
                Log::info("        src [${column}] sum: $src_sum, dst sum: $dst_sum match.");
            }

            push @diff_numerical_result, {
                column => $column,
                src_sum => $src_sum,
                dst_sum => $dst_sum,
            };
        }

        # 获取样例sql
        my $base_sql = sprintf(_getSQL($migration, $comid, "src"), join(", ", @character_columns), $migration->{src_conn}->{schema});
        my $sampleRowSql = $base_sql . " ORDER BY RANDOM() LIMIT 1";
        Log::info("    Sampled row SQL: $sampleRowSql");
        my $sampleRow = $src_dbh->selectrow_hashref($sampleRowSql);

        my $rows_sql = sprintf(_getSQL($migration, $comid, "src"), " COUNT(1) ", $migration->{src_conn}->{schema});
        my @rows_conditions;
        my $rows_count;
        for my $character_column (@character_columns) {
            my $value = $sampleRow->{$character_column};
            if (!defined $value || $value eq "") {
                push @rows_conditions, " ($character_column IS NULL OR $character_column='') ";
            } else {
                push @rows_conditions, " $character_column = '$value' ";
            }
            $rows_count = $src_dbh->selectrow_array($rows_sql . " AND " . join(" AND ", @rows_conditions));
            if ($rows_count <= 200) {
                last;
            }
        }

        my @diff_rows_result;
        if ($rows_count > 200) {
            Log::info("    Too many rows, skip character columns check.");
        } else {
            Log::info("    Simpled row count: ${rows_count}");

            $rows_sql = sprintf(_getSQL($migration, $comid, "src"), join(", ", @character_columns), $migration->{src_conn}->{schema})
                . " AND " . join(" AND ", @rows_conditions) . " ORDER BY " . join(" NULLS FIRST, ", @character_columns);
            Log::info("    Sampled rows SQL: ${rows_sql}");
            my $src_sth = $src_dbh->prepare($rows_sql);
            $src_sth->execute();
            my $src_rows = $src_sth->fetchall_arrayref({});
            
            $rows_sql = sprintf(_getSQL($migration, $comid, "dst"), join(", ", @character_columns), $coms{$comid}->{schema})
                . " AND " . join(" AND ", @rows_conditions) . " ORDER BY " . join(" NULLS FIRST, ", @character_columns);
            my $dst_sth = $dst_dbh->prepare($rows_sql);
            $dst_sth->execute();
            my $dst_rows = $dst_sth->fetchall_arrayref({});
            

            for my $i (0..$#$src_rows) {
                my $src_row = $src_rows->[$i];
                my $dst_row = $dst_rows->[$i];

                my @row_result;
                for my $character_column (@character_columns) {
                    my $src_value = $src_row->{$character_column};
                    my $dst_value = $dst_row->{$character_column};

                    unless (defined $src_value) {
                        $src_value = "";
                    }
                    unless (defined $dst_value) {
                        $dst_value = "";
                    }
                    if ($src_value ne $dst_value) {
                        Log::error("        src [${character_column}] value: $src_value, dst value: $dst_value miss match!");
                    } else {
                        Log::info("        src [${character_column}] value: $src_value, dst value: $dst_value match.");
                    }

                    push(@row_result, (
                        $src_value, $dst_value, $src_value eq $dst_value? "是" : "否",
                    ));
                }
                push @diff_rows_result, \@row_result;
            }
        }
        

        my $migration_name = basename($migration_file);
        $migration_name =~ s/\.json$//;
        # 按分省导出报表
        my $excel = new Export(
            filename => "${comid}.xlsx",
            subdir => "fensheng-report/${migration_name}/" . strftime("%Y%m%d", localtime),
            time => '.',
        );

        $excel->addWorksheet("行数对比", [
            "表名",
            "原表行数",
            "目标表行数",
            "是否一致",
        ]);
        $excel->addWorksheetData("行数对比", [
            $migration->{table},
            $src_count,
            $dst_count,
            $src_count == $dst_count? "是" : "否",
        ]);
        $excel->doExport("行数对比");

        $excel->addWorksheet("求和对比", [
            "表名",
            "列名",
            "原表列求和结果",
            "目标表列求和结果",
            "是否一致",
        ]);
        for my $i (0..$#diff_numerical_result) {
            my $row = $i + 1;
            $excel->addWorksheetData("求和对比", [
                $migration->{table},
                $diff_numerical_result[$i]->{column},
                $diff_numerical_result[$i]->{src_sum},
                $diff_numerical_result[$i]->{dst_sum},
                $diff_numerical_result[$i]->{src_sum} == $diff_numerical_result[$i]->{dst_sum}? "是" : "否",
            ]);
        }
        $excel->doExport("求和对比");

        my @col_sheet_headers = (
            "表名",
        );
        for my $i (0..$#character_columns) {
            push @col_sheet_headers, "原表列[${character_columns[$i]}]";
            push @col_sheet_headers, "目标表列[${character_columns[$i]}]";
            push @col_sheet_headers, "列[${character_columns[$i]}]是否一致";
        }
        $excel->addWorksheet("随机行列对比", \@col_sheet_headers);
        for my $i (0..$#diff_rows_result) {
            my $row = $i + 1;
            my @row_data = (
                $migration->{table},
            );
            for my $j (0..$#{$diff_rows_result[$i]}) {
                push @row_data, $diff_rows_result[$i][$j];
            }
            $excel->addWorksheetData("随机行列对比", \@row_data);
        }
        $excel->doExport("随机行列对比");

        $excel->close();
        Log::info("    > Export to excel: ${comid}.xlsx");
    }
}

sub genMain
{
    my $time = shift;
    my $gen_list = shift;

    unless (defined $time) {
        Log::error("time is required.");
        return;
    }

    my $migration_path = $Const::CONF_PATH. $Const::DS. "migrations-fen";
    my %migrations;
    find(sub {
        if (/\.json$/ && -f $_) {
            my $filename = $_;
            my $basename = basename($File::Find::name);
            my $migration_name = basename($basename, ".json");
            $migrations{$migration_name} = decode_json(Utils::readFile($basename));
        }
    }, $migration_path);

    my $coms_path = $Const::CONF_PATH. $Const::DS. "conns-fen";
    my %coms;
    find(sub {
        if (/\.json$/ && -f $_) {
            my $filename = $_;
            my $basename = basename($File::Find::name);
            my $comid = basename($basename, ".json");
            $coms{$comid} = decode_json(Utils::readFile($basename));
        }
    }, $coms_path);

    generatedMainExcel(\%migrations, \%coms, $time);
}

sub genMain2 {
    my $dir = shift;
    my $migration = shift;
    unless (defined $dir) {
        Log::error('missing argument $dir');
        return 1;
    }

    unless (-d $dir) {
        Log::error("Not found dir: ${dir}");
        return 1;
    }

    Log::info("\$dir is: ${dir}");
    if (defined $migration) {
        Log::info("zhiding migration: $migration");
    }

    opendir (my $dh, $dir) or die "Can not open dir ${dir}";
    while (my $entry = readdir($dh)) {
        next if $entry =~ /^\.\.?$/;
        my $path = "$dir/$entry";
        next if -f ($path);
        if (defined $migration) {
            next if $entry ne $migration;
        }
        Log::info("Table: ${entry}");
        opendir (my $ddh, $path) or die "Can not open dir ${path}";
        
        my $excel_filename = "0_main.xlsx";
        my $excel = new Export(filename => $excel_filename, subdir => "fensheng-report/${entry}", time => '.');
        $excel->addWorksheet("总览",["日期","comid", "行数是否一致", "列求和是否一致", "随机行列是否一致", "结果链接"]);
        while (my $dentry = readdir($ddh)) {
            next if $dentry =~ /^\.\.?$/;
            my $dpath = "$path/$dentry";
            next if -f ($dpath);
            Log::info("  Date: $dentry");
            
            opendir (my $cdh, $dpath) or die "Con not open dir ${dpath}";
            while (my $centry = readdir($cdh)) {
                next if $centry =~ /^\.\.?$/;
                my $cpath = "$dpath/$centry";
                my $comid = $cpath;
                $comid =~ s/\.xlsx$//;
                Log::info("    Comid: $centry");

                my ($diff_count, $diff_sum, $diff_row) = (1,1,1);
                my $parser = Spreadsheet::ParseXLSX->new();
                my $workbook = $parser->parse($cpath);
                for my $worksheet ($workbook->worksheets()) {
                    my $worksheet_name = $worksheet->get_name();
                    if ($worksheet_name eq "行数对比") {
                        $diff_count = $worksheet->get_cell(1, 3)->value() eq "是"? 1 : 0;
                    } elsif ($worksheet_name eq "求和对比") {
                        my $row_max = $worksheet->row_range();
                        for my $row (1..$row_max) {
                            my $cell = $worksheet->get_cell($row, 4);
                            if ($cell->value() eq "否") {
                                $diff_sum = 0;
                                last;
                            }
                        }
                    } elsif ($worksheet_name eq "随机行列对比") {
                        my $row_max = $worksheet->row_range();
	                my $col_max = $worksheet->col_range();
                        for my $row (1..$row_max) {
                            for (my $col_i = 3; $col_i < $col_max; $col_i = $col_i + 3) {
                                my $cell = $worksheet->get_cell($row, $col_i);
                                if ($cell && $cell->value() ne "是") {
                                    $diff_row = 0;
                                    last;
                                }
                            }
                        }
                    }
                }
                $excel->addWorksheetData("总览", [
                    $dentry,
                    $comid,
                    $diff_count == 1? "是" : "否",
                    $diff_sum == 1? "是" : "否",
                    $diff_row == 1? "是" : "否",
                    "${dentry}/${centry}",
                ]);
                Log::info("Excel for table ${dentry} generated.");
                    
            }
            closedir $cdh;
        }
        closedir $ddh;
        $excel->doExport("总览");
        Log::primary("Excel for 0_main generated at [" . $excel->getRealpath(). "].");
        $excel->close();
        undef $excel;
    }
    closedir $dh;
}

sub generatedMainExcel
{
    my ($migrations, $coms, $time) = @_;

    Log::info("Genrated validation main excel.");

    my $filename = "0_main" . strftime("%Y%m%d", localtime) . ".xlsx";

    my $excel = new Export(
        filename => $filename,
        subdir => "fen",
        time => $time,
    );

    $excel->addWorksheet("总览", [
        "表名",
        "行数是否一致",
        "列求和是否一致",
        "随机行列是否一致",
        "结果链接",
        "备注",
    ]);
    
    my @all_main_data;
    while (my ($migration_name, $migration) = each %$migrations) {
        Log::info("Generating main excel for migration [ ${migration_name} ].");


        for my $comid (keys %$coms) {
            my ($diff_count, $diff_sum, $diff_row) = (1, 1, 1);

            my $parser = Spreadsheet::ParseXLSX->new();
            my $excel_filename = $Const::DATA_PATH . $Const::DS . $time. $Const::DS. "fen". $Const::DS . $migration_name. $Const::DS. "${comid}.xlsx";

            if (! -e $excel_filename) {
                Log::warn("Excel file not found: $excel_filename");
                next;
            }

            my $workbook = $parser->parse($excel_filename);
            for my $worksheet ($workbook->worksheets()) {
                my $worksheet_name = $worksheet->get_name();

                if ($worksheet_name eq "行数对比") {
                    $diff_count = $worksheet->get_cell(1, 3)->value() eq "是"? 1 : 0;
                } elsif ($worksheet_name eq "求和对比") {
                    my $row_max = $worksheet->row_range();
                    for my $row (1..$row_max) {
                        my $cell = $worksheet->get_cell($row, 4);
                        if ($cell->value() eq "否") {
                            $diff_sum = 0;
                            last;
                        }
                    }
                } elsif ($worksheet_name eq "随机行列对比") {
                    my $row_max = $worksheet->row_range();
                    my $col_max = $worksheet->col_range();
                    for my $row (1..$row_max) {
                        for (my $col_i = 3; $col_i < $col_max; $col_i = $col_i + 3) {
                            my $cell = $worksheet->get_cell($row, $col_i);
                            if ($cell && $cell->value() ne "是") {
                                $diff_row = 0;
                                last;
                            }
                        }
                    }                  
                }
            }

            $excel->addWorksheetData("总览", [
                $migration->{table},
                $diff_count == 1? "是" : "否",
                $diff_sum == 1? "是" : "否",
                $diff_row == 1? "是" : "否",
                "${migration_name}/${comid}.xlsx",
            ]);
            Log::info("Excel for migration [ ${migration_name} ] [ ${comid} ] generated.");
        }
    }

    $excel->doExport("总览");
    Log::primary("Excel for 0_main generated at [". $excel->getRealpath(). "].");

    $excel->close();
    undef $excel;
}

sub getMaxProcesses
{
    my $app_config_path = $Const::CONF_PATH . $Const::DS . "app.config";
    my $app_config = decode_json(Utils::readFile($app_config_path));
    return defined $app_config->{migration}->{max_processes} ? $app_config->{migration}->{max_processes} : 5;
}

sub _getSQL
{
    my $migration = shift;
    my $comid = shift;
    my $conn_type = shift;

    my $sql;
    if ($migration->{sql} ne "") {
        $sql = $migration->{sql};
    } else {
        my $table = $migration->{table};
        if ($conn_type eq 'src' && defined $migration->{src_table} && $migration->{src_table} ne "") {
            $table = $migration->{src_table};
        }
        $sql = "SELECT %s FROM %s.$table";
    }

    if ($migration->{is_com_table} == 0 && defined $comid) {
        my $comid_column_name = (defined $migration->{comid_column_name} && $migration->{comid_column_name} ne "") ? $migration->{comid_column_name} : "comcode";
        if (defined $migration->{is_long_comid} && $migration->{is_long_comid} == 1) {
            $sql .= " WHERE (CASE WHEN SUBSTR(${comid_column_name},1,4) IN ('2102','3302','3502','3702','4403')  THEN SUBSTR(${comid_column_name},1,4) = '${comid}' ELSE SUBSTR(${comid_column_name},1,2) || '00' = '${comid}' END) ";
        } else {
            $sql .= " WHERE " . $comid_column_name . "  = '$comid' ";
        }
    } else {
        $sql.= " WHERE 1=1 ";
    }

    return $sql;
}

1;
